package org.example.server.search.listener.mq.kafka;

import cn.hutool.core.lang.TypeReference;
import cn.hutool.json.JSONUtil;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.Conflicts;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.InlineScript;
import co.elastic.clients.elasticsearch._types.Script;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch.core.UpdateByQueryRequest;
import co.elastic.clients.json.JsonData;
import jakarta.annotation.Resource;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.example.frw.canal.EntityComposite;
import org.example.frw.canal.HashValue;
import org.example.frw.canal.binlog.BaseCanalBinlog;
import org.example.frw.canal.binlog.entity.CategoryBinlogEntity;
import org.example.frw.canal.config.TopicConcurrentlySetting;
import org.example.frw.canal.listener.mq.kafka.BaseConsumer;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.BatchAcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Component
@Slf4j
public class CategoryConsumer extends BaseConsumer<CategoryBinlogEntity> implements BatchAcknowledgingMessageListener<String, String> {
    private static final String TOPIC = "goods__tb_category";
    private static final String GROUP = "search-server";

    @Resource
    protected ElasticsearchClient elasticsearchClient;


    public CategoryConsumer(TopicConcurrentlySetting topicConcurrentlySetting) {
        super(topicConcurrentlySetting, TOPIC,GROUP);
    }

    /**
     * 幂等由ES保证
     *
     * @param entityComposite entityComposite
     */
    @Override
    @SneakyThrows
    protected void doProcess(EntityComposite<? extends HashValue> entityComposite) {
        log.info("CategoryConsumer processing ,partition= {} entityComposite={}", entityComposite.getPartition(),entityComposite);
        CategoryBinlogEntity categoryEntity = (CategoryBinlogEntity) entityComposite.getEntity();
        switch (entityComposite.getType()) {
            case "UPDATE" -> {
                String scriptSource = """
                            ctx._source.category.categoryName = params.categoryName;
                            ctx._source.category.path = params.path;
                            ctx._source.category.isDisplay = params.isDisplay;
                            ctx._source.category.sort = params.sort;
                            ctx._source.category.unit = params.unit;
                            ctx._source.category.createTime = params.createTime;
                            ctx._source.category.updateTime = params.updateTime;
                        """;
                Map<String, JsonData> map = new HashMap<>();
                map.put("categoryName", JsonData.of(categoryEntity.getCategoryName()));
                map.put("path", JsonData.of(categoryEntity.getPath()));
                map.put("isDisplay", JsonData.of(String.valueOf(categoryEntity.getIsDisplay())));
                map.put("sort", JsonData.of(String.valueOf(categoryEntity.getSort())));
                map.put("unit", JsonData.of(categoryEntity.getUnit()));
                map.put("createTime", JsonData.of(categoryEntity.createTimeTimestamp()));
                map.put("updateTime", JsonData.of(categoryEntity.updateTimeTimestamp()));

                UpdateByQueryRequest request = UpdateByQueryRequest.of(
                        builder -> builder
                                .index("goods")
                                .query(Query.of(q -> q.nested(n ->
                                        n.path("category")
                                                .query(Query.of(p ->
                                                        p.bool(t ->
                                                                t.must(m ->
                                                                        m.match(mc ->
                                                                                mc.field("category.id").query(FieldValue.of(categoryEntity.getId())))))))
                                )))
                                .script(Script.of(s -> s.inline(
                                        new InlineScript.Builder().lang("painless").source(scriptSource).params(map).build()
                                )))
                                .conflicts(Conflicts.Proceed)
                );
                log.info("{}", request);
                elasticsearchClient.updateByQuery(request);

            }
            case "INSERT", "DELETE" -> {
                // ignore，不需要处理
            }
        }
        log.info("processing done！！！");
    }

    @Override
    protected List<EntityComposite<CategoryBinlogEntity>> hashIndex(int concurrentlyLevel, ConsumerRecord<String, String> messageExt) {
        log.info("再次分区，concurrentlyLevel={},key={},partition={},offset={}", concurrentlyLevel, messageExt.key(), messageExt.partition(), messageExt.offset());
        String body = messageExt.value();
        TypeReference<BaseCanalBinlog<CategoryBinlogEntity>> typeReference = new TypeReference<>() {
        };
        BaseCanalBinlog<CategoryBinlogEntity> bean = JSONUtil.toBean(body, typeReference, true);
        return getEntityComposites(concurrentlyLevel, bean);
    }


    @Override
    @KafkaListener(topics = TOPIC, groupId = GROUP, containerFactory = BATCH_CONTAINER_FACTORY)
    public void onMessage(List<ConsumerRecord<String, String>> records, Acknowledgment acknowledgment) {
        super.consumeMessage(records, acknowledgment);
    }
}